package com.jeffse.light.stream.core.data.operator.Covert;

import com.jeffse.light.stream.core.data.descriptor.operatorDescriptor.OperatorDescriptor;
import com.jeffse.light.stream.core.function.base.KeyedProcessFunction;

public class KeyedProcessConvert<T> implements BaseConvert<T> {
    @Override
    public void convert(OperatorDescriptor operatorDescriptor, Object value) throws Exception {
        KeyedProcessFunction keyedProcessFunction =  (KeyedProcessFunction)operatorDescriptor.getFunc();
        synchronized(keyedProcessFunction){
            keyedProcessFunction.setCurrentKey(operatorDescriptor.getKeyedSelector().keyBy(value));
            keyedProcessFunction.setNextOperatorDescriptor(operatorDescriptor.getNetOperatorDescriptor());

            // event类型必须设置水印才能触发
            if(operatorDescriptor.getWaterMarkFunction() != null){
                keyedProcessFunction.WaterMark(operatorDescriptor.getWaterMarkFunction().setWaterMark(value));
                keyedProcessFunction.triggerEvent();
            }
            keyedProcessFunction.processElement(keyedProcessFunction, value);
        }
    }
}
